package gnss

import (
	"container/list"
	"dgo/goutils/utils"
	"encoding/json"
	"fmt"
	"log"
	"net"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

const (
	TYPE_UNKOWN      = 0
	TYPE_NTRIPSERVER = 1
	TYPE_NTRIPCLIENT = 2
	BOOL_TRUE        = 1
	BOOL_FALSE       = -1

	CLIENT_AUTH_ERR_PASS         = -2
	CLIENT_AUTH_NONE_PERMISSIONS = -3

	ICY_200_OK         = "ICY 200 OK\r\n\r\n"
	ERR_BADPASS        = "ERROR - Bad Password\r\n"
	NTRIP_UNAUTHORIZED = "HTTP/1.0 401 Unauthorized"
	ERR_INNER          = "ERROR - Inner error \r\n"
	ERR_BADMountPoint  = "ERROR - Bad MountPoint\r\n"
)

type NtripServerInfo struct {
	MountPoint      string
	RecvBytes       int64
	ConnectCounter  int32
	LastRcvTime     string
	LastConnectTime string
	FromAddr        string
	ClientCounter   int
}

type CasterNtripClient struct {
	ID          string
	Pass        string
	MountPoints string // * 代表所有, "RTKGG;RTKBD;"
}

type CasterNtripServer struct {
	ctx            *NtripContext
	srctext        string
	consumers      *list.List
	consumersLock  *sync.RWMutex
	lock           *sync.RWMutex
	fromAddr       string
	connectcounter int32
	recvbytes      int64

	lastConnectTime time.Time
	lastRecvTime    time.Time
}

// caster
type NtripCaster struct {
	isTerminated          bool
	OnlineCount           int32 // 在线数量
	requestcounter        int64 // 处理请求数量
	sourcetableBuf        []byte
	sourcetableSvrList    *list.List
	NtripSvrMap           map[string]*CasterNtripServer // ntripserver 管理Map
	NtripClientMap        map[string]*CasterNtripClient // ntripclient 管理Map
	OnlineContexts        map[uint32]*NtripContext      // 在线列表
	onlinelocker          *sync.RWMutex
	MaxSendQueueSize      uint16
	lock                  *sync.RWMutex
	AllowUnkownMountPoint byte   //是否允许未知的源节点接入
	NtripserverAuthstr    string // 接入密码
}

// 连接handle
var _contexthandle uint32 = 1

type NtripContextInfo struct {
	FromAddr      string
	RecvBytes     int64
	SendBytes     int64
	ConnectedTime string
	MountPoint    string
	User          string
}

// 连接对象
type NtripContext struct {
	ContextHandle uint32
	Type          uint8 // 连接对象类型， CasterNtripServer, CasterNtripClientMap
	caster        *NtripCaster
	conn          net.Conn
	requestParser *NtripRequest
	user          string
	sendQueue     chan interface{}
	closeWaitg    sync.WaitGroup
	_NtripSvr     *CasterNtripServer
	_Lock         *sync.RWMutex
	closeFlag     int32
	recvbytes     int64
	sendbytes     int64
	connecttime   time.Time
	lastsendtime  time.Time
	lastrecvtime  time.Time
}

func newNtripServer() *CasterNtripServer {
	return &CasterNtripServer{
		ctx:           nil,
		lock:          new(sync.RWMutex),
		consumers:     list.New(),
		consumersLock: new(sync.RWMutex),
	}
}

func NewNtripCaster() *NtripCaster {
	return &NtripCaster{
		isTerminated:       false,
		OnlineCount:        0,
		lock:               new(sync.RWMutex),
		onlinelocker:       new(sync.RWMutex),
		sourcetableSvrList: list.New(),
		NtripSvrMap:        map[string]*CasterNtripServer{},
		NtripClientMap:     map[string]*CasterNtripClient{},
		OnlineContexts:     make(map[uint32]*NtripContext),
		NtripserverAuthstr: "0000",
		MaxSendQueueSize:   1024,
	}
}

func newNtripContext(_conn net.Conn, caster *NtripCaster) *NtripContext {
	return &NtripContext{
		ContextHandle: atomic.AddUint32(&_contexthandle, 1),
		conn:          _conn,
		caster:        caster,
		sendQueue:     make(chan interface{}, caster.MaxSendQueueSize),
		requestParser: NewNtripRequest(),
		_Lock:         new(sync.RWMutex),
		_NtripSvr:     nil,
		lastrecvtime:  time.Now(),
		lastsendtime:  time.Now(),
	}
}

// 添加一个NtripServer
func (self *NtripCaster) CheckConfigNtripServer(id string, str string) {
	self.lock.Lock()
	self.innerCheckAddNtripServer(id, str)
	self.lock.Unlock()
	self.RebuildSourceTableBuffer()
}

// 检测死链接
func (self *NtripCaster) CheckKickOut() {
checkfor:
	for {
		if self.isTerminated {
			break checkfor
		}
		time.Sleep(time.Second * 20)

		var lst []*NtripContext
		lst = self.CopyOnlineContexts()
		for _, ctx := range lst {
			ctx.checkKickOut(20)
		}
	}
}

// 添加一个NtripServer
func (self *NtripCaster) innerCheckAddNtripServer(id string, str string) {
	var svr = self.NtripSvrMap[id]
	if svr == nil {
		svr = newNtripServer()
		svr.srctext = str
		self.NtripSvrMap[id] = svr
		self.sourcetableSvrList.PushBack(id)
	} else {
		svr.srctext = str
	}
}

// 添加一个NtripServer
func (_self *NtripCaster) checkAutoWireNtripServer(id string) *CasterNtripServer {
	flag := 0
	_self.lock.Lock()
	svr := _self.NtripSvrMap[id]
	if svr == nil {
		svr = newNtripServer()
		str := fmt.Sprintf("STR;%s;%s;UNKOWN;UNKOWN;0;UNKOWN;misc;CHN;0;0;1;1;DiocpVRS;none;B;N;0;DiocpGW;DiocpGW;", id, id)
		svr.srctext = str
		_self.NtripSvrMap[id] = svr
		_self.sourcetableSvrList.PushBack(id)
		flag = 1
	}
	_self.lock.Unlock()

	if flag == 1 {
		_self.RebuildSourceTableBuffer()
	}
	return svr
}

// 判断是否存在某个NtripServer
func (_self *NtripCaster) CheckHasNtripServer(id string) bool {
	_self.lock.RLock()
	r := _self.NtripSvrMap[id] == nil
	_self.lock.RUnlock()
	return r
}

func (self *NtripCaster) CopyOnlineContexts() []*NtripContext {
	self.lock.RLock()
	rval := make([]*NtripContext, 0, len(self.OnlineContexts))
	for _, v := range self.OnlineContexts {
		rval = append(rval, v)
	}
	self.lock.RUnlock()
	return rval
}

func (self *NtripCaster) CopyOnlineNtripClientsContexts() []*NtripContext {
	self.lock.RLock()
	rval := make([]*NtripContext, 0, len(self.OnlineContexts))
	for _, v := range self.OnlineContexts {
		if v.Type == TYPE_NTRIPCLIENT {
			rval = append(rval, v)
		}
	}
	self.lock.RUnlock()
	return rval
}

// 获取某个NtripServer
func (_self *NtripCaster) CheckGetNtripServer(id string) *CasterNtripServer {
	_self.lock.RLock()
	r := _self.NtripSvrMap[id]
	_self.lock.RUnlock()
	return r
}

func (_self *NtripCaster) SetAllowUnkownMountPoint(v byte) {
	_self.AllowUnkownMountPoint = v
}

func (_self *NtripCaster) SetNtripServerAuthStr(ntripsvrAuthStr string) {
	_self.NtripserverAuthstr = ntripsvrAuthStr
}

func (self *NtripCaster) SetNtripClient(id string, pass string, accessMountPoints string) {
	self.lock.Lock()
	itm := self.NtripClientMap[id]
	if itm == nil {
		itm = &CasterNtripClient{
			ID:          id,
			Pass:        pass,
			MountPoints: accessMountPoints,
		}
		self.NtripClientMap[id] = itm
	} else {
		itm.Pass = pass
		if len(accessMountPoints) > 0 {
			itm.MountPoints = accessMountPoints
		}
	}
	self.lock.Unlock()
}

func (self *NtripCaster) ResponseClientSourceTable(ctx *NtripContext) {
	req := ctx.requestParser
	id, pass, err := req.DecodeNtripClientAuthInfo()
	if err != nil {
		log.Println(ctx.conn.RemoteAddr().String(), " 获取认证信息出现了异常:", err)
	}

	ms := ""

	self.lock.RLock()
	defer self.lock.RUnlock()
	if len(id) > 0 {
		itm := self.NtripClientMap[id]
		if itm != nil {
			if itm.Pass == pass {
				ms = itm.MountPoints
				if len(ms) == 0 {
					ms = "NONE"
				}
			}
		}
	}

	if len(ms) == 0 {
		// 如果guest也没有就没有源节点
		ms = "NONE"
		itm := self.NtripClientMap["guest"]
		if itm != nil {
			ms = itm.MountPoints
		}
	}
	var postBuf []byte

	if ms == "*" {
		postBuf = self.sourcetableBuf
	} else {
		msarr := strings.Split(ms, ";")
		stables := ""
		for _, v := range msarr {
			sitm := self.NtripSvrMap[v]
			if sitm != nil {
				stables += sitm.srctext + "\r\n"
			}
		}
		if len(stables) == 0 {
			log.Println(ctx.conn.RemoteAddr().String(), " 对应指定的源节点无效", ms)
		} else {
			postBuf = BuildSourceTableBuffer(stables)
		}
	}
	ctx.PostSendBuffer(postBuf)
	// ctx.conn.Write(postBuf)
}

func (self *NtripCaster) CheckClientAuth(ctx *NtripContext) int {
	req := ctx.requestParser
	id, pass, err := req.DecodeNtripClientAuthInfo()
	if err != nil {
		log.Println(ctx.conn.RemoteAddr().String(), " 获取认证信息出现了异常")
	}

	if len(id) == 0 {
		id = "guest"
	}

	ctx.user = id

	rval := CLIENT_AUTH_ERR_PASS
	self.lock.RLock()
	itm := self.NtripClientMap[id]
	if itm != nil {
		if strings.EqualFold(id, "guest") || itm.Pass == pass {
			if itm.MountPoints == "*" {
				rval = BOOL_TRUE
			} else {
				s := req.MountPoint + ";"
				if strings.Contains(itm.MountPoints, s) {
					rval = BOOL_TRUE
				} else {
					rval = CLIENT_AUTH_NONE_PERMISSIONS
					log.Println(ctx.conn.RemoteAddr().String(), " 源节点访问权限不足,用户名:", id, "请求访问源节点：", req.MountPoint,
						"可访问源节点:", itm.MountPoints)
				}
			}
		} else {
			rval = CLIENT_AUTH_ERR_PASS
			log.Println(ctx.conn.RemoteAddr().String(), " 认证失败,用户名:", id, "密码:", pass)
		}
	}
	self.lock.RUnlock()

	return rval

}

func (_self *NtripCaster) RebuildSourceTableBuffer() {

	str := ""
	_self.lock.RLock()
	l := _self.sourcetableSvrList
	for e := l.Front(); e != nil; e = e.Next() {
		v := e.Value.(string)
		itm := _self.NtripSvrMap[v]
		if itm != nil {
			str += itm.srctext + "\r\n"
		}
	}
	_self.lock.RUnlock()

	buf := BuildSourceTableBuffer(str)

	_self.lock.Lock()
	_self.sourcetableBuf = buf
	_self.lock.Unlock()
}

func (_self *NtripCaster) getSourceTableBuf() []byte {
	_self.lock.RLock()
	r := _self.sourcetableBuf
	_self.lock.RUnlock()
	return r
}

func (_self *NtripCaster) LoadNtripSvrFromSourceTableText(str string) {
	strs := strings.Split(str, "\n")
	_self.lock.Lock()
	for _, v := range strs {
		itm := strings.SplitN(v, ";", 3)
		if len(itm) == 3 {
			_self.innerCheckAddNtripServer(itm[1], v)
		} else {
			log.Println(" 错误的SourceTable格式:", v)
		}
	}
	_self.lock.Unlock()
	_self.RebuildSourceTableBuffer()
}

func (self *NtripCaster) addOnline(ctx *NtripContext) {
	self.onlinelocker.Lock()
	self.OnlineContexts[ctx.ContextHandle] = ctx
	self.onlinelocker.Unlock()
	atomic.AddInt32(&self.OnlineCount, 1)
}

func (self *NtripCaster) removeOnline(ctx *NtripContext) {
	self.onlinelocker.Lock()
	delete(self.OnlineContexts, ctx.ContextHandle)
	self.onlinelocker.Unlock()
	NN := -1
	atomic.AddInt32(&self.OnlineCount, ^int32(-NN-1))
}

// 开启一个侦听
func (_self *NtripCaster) Start(listenAddress string, port int) {

	_addr := fmt.Sprintf("%s:%d", listenAddress, port)
	Log("Caster 侦听地址:", _addr)
	Log("Caster NtripServer接入密码:", _self.NtripserverAuthstr)
	netListen, err := net.Listen("tcp", _addr)
	if err != nil {
		Log("侦听出现异常:", err)
		return
	}

	defer netListen.Close()

	_self.isTerminated = false

	Log("等待连接")
	for {
		_conn, err := netListen.Accept()
		if err != nil {
			continue
		}
		ctx := newNtripContext(_conn, _self)
		ctx.connecttime = time.Now()
		_self.addOnline(ctx)
		ctx.StartHandleTask()
		utils.GoFunc2(func() {
			handleConnection(_self, ctx)
		})
	}

}

// 查询NtripServer信息
func (_self *NtripCaster) QueryNtripServerInfo() *list.List {
	rlst := list.New()
	_self.lock.RLock()
	l := _self.sourcetableSvrList
	for e := l.Front(); e != nil; e = e.Next() {
		id := e.Value.(string)
		v := _self.NtripSvrMap[id]
		if v != nil {
			itm := &NtripServerInfo{
				MountPoint:      id,
				FromAddr:        v.fromAddr,
				RecvBytes:       v.recvbytes,
				LastConnectTime: StringTime(v.lastConnectTime),
				LastRcvTime:     StringTime(v.lastRecvTime),
				ClientCounter:   v.consumers.Len(),
			}
			rlst.PushBack(itm)
		}
	}
	_self.lock.RUnlock()
	return rlst
}

// 查询NtripServer信息
func (self *NtripCaster) QueryNtripClientContext(pos int) []*NtripContextInfo {
	lst := self.CopyOnlineNtripClientsContexts()

	rval := make([]*NtripContextInfo, 0, 1000)
	if len(lst) < pos {
		return rval
	}
	j := 0
	for _, v := range lst[pos:] {
		itm := &NtripContextInfo{
			FromAddr:      v.conn.RemoteAddr().String(),
			ConnectedTime: StringTime(v.connecttime),
			RecvBytes:     v.recvbytes,
			SendBytes:     v.sendbytes,
			MountPoint:    v.requestParser.MountPoint,
			User:          v.user,
		}
		rval = append(rval, itm)
		j++
		if j >= 1000 {
			break
		}
	}
	return rval
}

// 查询NtripServer信息
func (_self *NtripCaster) QueryNtripServerInfoList() []*NtripServerInfo {

	_self.lock.RLock()

	l := _self.sourcetableSvrList
	rval := make([]*NtripServerInfo, 0, l.Len())
	j := 0
	for e := l.Front(); e != nil; e = e.Next() {
		id := e.Value.(string)
		v := _self.NtripSvrMap[id]
		if v != nil {
			itm := &NtripServerInfo{
				MountPoint:      id,
				FromAddr:        v.fromAddr,
				RecvBytes:       v.recvbytes,
				LastConnectTime: StringTime(v.lastConnectTime),
				LastRcvTime:     StringTime(v.lastRecvTime),
				ClientCounter:   v.consumers.Len(),
			}
			rval = append(rval, itm)
			j++
		}
	}
	_self.lock.RUnlock()
	return rval
}

// 处理连接
func handleConnection(_self *NtripCaster, ctx *NtripContext) {
	// Log(ctx.conn.RemoteAddr().StatusString(), " tcp connect success")

	buffer := make([]byte, 2048)

	for {
		n, err := ctx.conn.Read(buffer)
		if err != nil {
			Log(ctx.conn.RemoteAddr().String(), " 处理接受请求时出现了异常: ", err)
			ctx.RequestDisconnect()
			return
		}

		ctx.lastrecvtime = time.Now()
		ctx.recvbytes = ctx.recvbytes + int64(n)
		if ctx.Type == TYPE_NTRIPSERVER {
			// Log(ctx.conn.RemoteAddr().StatusString(), " NTRIP SERVER DATA IN :", n)
			ctx.ProcessNtripServerBuffer(buffer[:n])
			//ctx.conn.Write(buffer[:n])
		} else {
			var r int
			nr := ctx.requestParser
			for _, data := range buffer {
				r = nr.InputBuffer(data)
				if r == DECODE_HEAD {
					if nr.DecodeHeader() == TYPE_UNKOWN {
						ctx.RequestDisconnect()
						return
					}

					ctx.Type = nr.GetRequestType()

					if ctx.Type == TYPE_NTRIPSERVER {
						if ctx.ResponseNtripServerRequest() == -1 {
							return
						}
						continue
					}

					if ctx.Type == TYPE_NTRIPCLIENT {
						if ctx.ResponseNtripClientHeaderRequest() == BOOL_FALSE {
							ctx.RequestDisconnect()
							return
						}
					}

					//Log(ctx.conn.RemoteAddr().StatusString(), " GetSourceTable ")
				} else if r == DECODE_NMEA {
					fmt.Printf("r:%d, size:%d, str:%s\n", r, nr.GetSize(), nr.GetBodyString())
				}
			}
		}

	}
}

func Log(v ...interface{}) {
	log.Println(v...)
}

func (self *NtripContext) DoClose() {
	if self.Type == TYPE_NTRIPSERVER {
		self.CheckUnSetAsNtripSvr()
	} else if self.Type == TYPE_NTRIPCLIENT {
		if self._NtripSvr != nil {
			self._NtripSvr.RemoveConsumer(self)
		}
	}
}

func (self *NtripContext) checkKickOut(isec int) {
	if self.Type == TYPE_NTRIPSERVER {
		t := time.Since(self.lastrecvtime)
		tick := int(t.Seconds())
		if tick > isec {
			log.Println(self.conn.RemoteAddr().String(), " 未接收数据时间:", tick, "秒，请求断开")
			self.RequestDisconnect()
		}
	} else if self.Type == TYPE_NTRIPCLIENT {
		t := time.Since(self.lastsendtime)
		tick := int(t.Seconds())
		if tick > isec {
			log.Println(self.conn.RemoteAddr().String(), " 未发送数据时间:", tick, "秒，请求断开")
			self.RequestDisconnect()
		}
	}
}

func (self *NtripContext) StartHandleTask() {
	// log.Println(self.conn.RemoteAddr().StatusString(), " 进入处理发送任务")
	self.closeWaitg.Add(1)
	utils.GoFunc2(self.HandleSend)
}

func (self *NtripContext) HandleSend() {
sendfor:
	for {
		select {
		case data, ok := <-self.sendQueue:
			if ok && data != nil {
				n, err := self.conn.Write(data.([]byte))
				if err != nil {
					Log(self.conn.RemoteAddr().String(), " 处理发送数据时出现了异常:", err)
				} else {
					self.lastsendtime = time.Now()
					self.sendbytes = self.sendbytes + int64(n)
					//log.Println(self.conn.RemoteAddr().StatusString(), " 数据发送完成", data)
				}
			} else {
				//log.Println(self.conn.RemoteAddr().StatusString(), " 收到nil数据,即将退出任务:")
				break sendfor
			}
		}
	}

	// Log(self.conn.RemoteAddr().StatusString(), " 发送任务退出！closeWaitg.Done()")
	self.closeWaitg.Done()
}

func (self *NtripContext) RequestDisconnect() {

	// 避免重复关闭
	if !atomic.CompareAndSwapInt32(&self.closeFlag, 0, 1) {
		return
	}

	// 投递一个空任务，让发送任务退出
	// log.Println(self.conn.RemoteAddr().StatusString(), " 投递空数据到队列(RequestDisconnect)")
	self.sendQueue <- nil

	conn := self.conn
	self.DoClose()
	// log.Println(self.conn.RemoteAddr().StatusString(), " 等待可以关闭")
	self.closeWaitg.Wait()

	// Log(conn.RemoteAddr().StatusString(), " 关闭连接(RequestDisconnect)")
	conn.Close()
	Log(conn.RemoteAddr().String(), " 请求关闭连接(RequestDisconnect), 当前连接数:", self.caster.OnlineCount)
	self.caster.removeOnline(self)
}

// 设定做为NtripServer连接
func (_self *NtripContext) CheckUnSetAsNtripSvr() {
	if _self._NtripSvr != nil {
		ntripsvr := _self._NtripSvr
		_self._Lock.Lock()
		_self._NtripSvr = nil
		_self._Lock.Unlock()
		log.Println(_self.conn.RemoteAddr().String(), " 取消绑定NtripSvr")
		ntripsvr.CheckUnBind(_self)
	}
}

// 解除作为NtripServer连接
func (self *NtripContext) SetAsNtripSvr(ntripSvr *CasterNtripServer) {
	self._Lock.Lock()
	self._NtripSvr = ntripSvr
	self._Lock.Unlock()
}

func (self *NtripContext) ResponseNtripClientHeaderRequest() int {

	if len(self.requestParser.MountPoint) == 0 {
		self.caster.ResponseClientSourceTable(self)
		return BOOL_FALSE
	}

	r := self.caster.CheckClientAuth(self)
	if r == BOOL_TRUE {
		self._NtripSvr = self.caster.CheckGetNtripServer(self.requestParser.MountPoint)
		if self._NtripSvr == nil {
			self.caster.ResponseClientSourceTable(self)
			return BOOL_FALSE
		}
		// 添加Ntrip的消费者
		self._NtripSvr.AddConsumer(self)

		self.PostSendBuffer([]byte(ICY_200_OK))
	} else if r == CLIENT_AUTH_ERR_PASS {
		self.conn.Write([]byte(NTRIP_UNAUTHORIZED))
		r = BOOL_FALSE
	} else if r == CLIENT_AUTH_NONE_PERMISSIONS {
		self.conn.Write([]byte(NTRIP_UNAUTHORIZED))
		r = BOOL_FALSE
	} else {
		self.conn.Write([]byte(NTRIP_UNAUTHORIZED))
		r = BOOL_FALSE
	}

	return r
}

func (_self *NtripContext) ProcessNtripServerBuffer(buffer []byte) int {
	if _self._NtripSvr != nil {
		_self._NtripSvr.recvbytes = _self._NtripSvr.recvbytes + int64(len(buffer))
		_self._NtripSvr.lastRecvTime = time.Now()

		_self._NtripSvr.DispatchBuffer(buffer)
	} else {
		Log(_self.conn.RemoteAddr().String(), " NO BINDING NTRIP SERVER")
	}
	return 0
}

// 响应一个NtripServer的请求
func (_self *NtripContext) ResponseNtripServerRequest() int {
	if _self.requestParser.AuthStr != _self.caster.NtripserverAuthstr {
		Log(_self.conn.RemoteAddr().String(), " NTRIP SERVER BAD Pass :", _self.requestParser.GetHeaderString())
		_self.PostSendBuffer([]byte(ERR_BADPASS))
		_self.RequestDisconnect()
		return -1
	}

	ntripsvr := _self.caster.CheckGetNtripServer(_self.requestParser.MountPoint)

	if ntripsvr == nil {
		Log(_self.conn.RemoteAddr().String(), " NTRIP SERVER NO MountPoint Config :", _self.requestParser.GetHeaderString())
		if _self.caster.AllowUnkownMountPoint == BOOL_TRUE {
			Log(_self.conn.RemoteAddr().String(), " NTRIP SERVER NO MountPoint Config, AutoWire :", _self.requestParser.GetHeaderString())
			ntripsvr = _self.caster.checkAutoWireNtripServer(_self.requestParser.MountPoint)
		} else {
			_self.PostSendBuffer([]byte(ERR_BADPASS))
			_self.RequestDisconnect()
			return -1
		}
	}

	// 踢出之前的连接,并解绑
	ntripsvr.CheckRequestDisconnect(_self)

	// 解绑之前的NtripSvr
	_self.CheckUnSetAsNtripSvr()

	// 绑定新的NtripSvr
	_self.SetAsNtripSvr(ntripsvr)

	// NtripSvr绑定新的Ctx
	ntripsvr.BindCtx(_self)

	_self.PostSendBuffer([]byte(ICY_200_OK))
	Log(_self.conn.RemoteAddr().String(), " NTRIP SERVER REQUEST IN :", _self.requestParser.GetHeaderString())
	return 0

}

func (self *NtripContext) PostSendBuffer(buffer []byte) {
	self.sendQueue <- buffer
	// log.Println(self.conn.RemoteAddr().StatusString(), " 投递发送任务,", buffer)
	// time.Sleep(time.Second)
}

///////////////////////////////////////////////////////////////////

// / 判断绑定的是否是ctx，如果不是则请求断开连接并解绑
func (self *CasterNtripServer) CheckRequestDisconnect(ctx *NtripContext) {

	self.lock.Lock()
	var _ctx *NtripContext = nil
	if self.ctx != ctx && self.ctx != nil {
		_ctx = self.ctx

		// 解除关系
		self.ctx = nil
	}
	self.lock.Unlock()

	if _ctx != nil {
		utils.GoFunc2(_ctx.RequestDisconnect)
	}
}

// / 判断绑定的是否是ctx，如果是则解绑
func (self *CasterNtripServer) CheckUnBind(ctx *NtripContext) {
	self.lock.Lock()
	if self.ctx == ctx {
		self.ctx = nil
	}
	self.lock.Unlock()
}

// / 绑定新的Ctx
func (self *CasterNtripServer) BindCtx(ctx *NtripContext) {
	self.lock.Lock()
	self.ctx = ctx
	self.lastConnectTime = time.Now()
	self.fromAddr = ctx.conn.RemoteAddr().String()
	self.connectcounter += 1
	self.lock.Unlock()
}

// / 添加请求数据客户端
func (self *CasterNtripServer) AddConsumer(ctx *NtripContext) {
	self.consumersLock.Lock()
	self.consumers.PushBack(ctx)
	self.consumersLock.Unlock()
}

// / 移除请求数据客户端
func (self *CasterNtripServer) RemoveConsumer(ctx *NtripContext) {
	self.consumersLock.Lock()
	l := self.consumers
	for e := l.Front(); e != nil; {
		if e.Value.(*NtripContext) == ctx {
			l.Remove(e)
			break
		} else {
			e = e.Next()
		}
	}
	self.consumersLock.Unlock()
}

// / 分发数据
func (self *CasterNtripServer) DispatchBuffer(buf []byte) {
	self.consumersLock.RLock()
	l := self.consumers
	for e := l.Front(); e != nil; e = e.Next() {
		ctx := e.Value.(*NtripContext)
		ctx.PostSendBuffer(buf)
	}
	self.consumersLock.RUnlock()
}

// / 输出信息
func (self *NtripServerInfo) ToString() string {
	jsons, errs := json.Marshal(self) //转换成JSON返回的是byte[]
	if errs != nil {
		fmt.Println(errs.Error())
	}
	return string(jsons)
}

func StringTime(aTime time.Time) string {
	if aTime.IsZero() {
		return ""
	} else {
		return time.Time(aTime).Format("2006-01-02 15:04:05")
	}
}

func DeepCopyMap(value interface{}) interface{} {
	if valueMap, ok := value.(map[int]interface{}); ok {
		newMap := make(map[int]interface{})
		for k, v := range valueMap {
			newMap[k] = DeepCopyMap(v)
		}
		return newMap
	} else if valueMap, ok := value.(map[string]interface{}); ok {
		newMap := make(map[string]interface{})
		for k, v := range valueMap {
			newMap[k] = DeepCopyMap(v)
		}
		return newMap
	} else if valueSlice, ok := value.([]interface{}); ok {
		newSlice := make([]interface{}, len(valueSlice))
		for k, v := range valueSlice {
			newSlice[k] = DeepCopyMap(v)
		}
		return newSlice
	}

	return value
}
